package com.atguigu.bigdata.chapter10;

import com.atguigu.bigdata.bean.OrderEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Author lzc
 * @Date 2022/9/8 9:02
 */
public class Flink06_High_Project_Order {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        KeyedStream<OrderEvent, Long> stream = env
            .readTextFile("input/OrderLog.csv")
            .map(new MapFunction<String, OrderEvent>() {
                @Override
                public OrderEvent map(String line) throws Exception {
                    String[] data = line.split(",");
                    return new OrderEvent(
                        Long.valueOf(data[0]),
                        data[1],
                        data[2],
                        Long.parseLong(data[3]) * 1000  // 把s变成ms
                    );
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((event, ts) -> event.getEventTime())
            )
            .keyBy(OrderEvent::getOrderId);
    
        // create 和 pay
        Pattern<OrderEvent, OrderEvent> pattern = Pattern
            .<OrderEvent>begin("create", AfterMatchSkipStrategy.skipPastLastEvent())
            .where(new SimpleCondition<OrderEvent>() {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "create".equals(value.getEventType());
                }
            }).optional()
            .next("pay")
            .where(new SimpleCondition<OrderEvent>() {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "pay".equals(value.getEventType());
                }
            })
            .within(Time.minutes(30));
    
        PatternStream<OrderEvent> ps = CEP.pattern(stream, pattern);
    
        SingleOutputStreamOperator<String> normal = ps.process(new MyPsFunction());
    
        normal.print("normal");
        normal.getSideOutput(new OutputTag<String>("timeout") {}).print("timeout");
    
    
        env.execute();
    }
    
    public static class MyPsFunction extends PatternProcessFunction<OrderEvent, String>
        implements TimedOutPartialMatchHandler<OrderEvent>{
    
        @Override
        public void processMatch(Map<String, List<OrderEvent>> map,
                                 Context ctx,
                                 Collector<String> out) throws Exception {
            // 取出订单id
            // 正常情况: 1. create + pay 2. pay
            // 如果给create加上optional, 那么正常的情况中是有可能没有create, 只有pay
            
            // 找异常情况: "1. create + pay" 是正常情况不要了, 只要 "2. pay"
            if (!map.containsKey("create")) {
                Long orderId = map.get("pay").get(0).getOrderId();
                out.collect("订单: " + orderId + " 只有pay...");
            }
        }
    
        @Override
        public void processTimedOutMatch(Map<String, List<OrderEvent>> map,
                                         Context ctx) throws Exception {
            // 超时的里面只有create, 不会有pay
            Long orderId = map.get("create").get(0).getOrderId();
            
            // 当create没有等到py的时候, 两种可能: 1. 只有create 2. pay超时了...
            ctx.output(new OutputTag<String>("timeout") {}, "订单: " + orderId + " 只有create或者pay超时支付...");
        }
    }
}
/*
1. 按照订单id keyBy

2. 对数据做处理
    create
        如果是create, 把他存入到状态

    pay
        如果是pay, 判断状态中是否有create,
        
        1. 如果没有create, 证明没有create, 只有pay
        2. 如果有create, 判断下pay的时间和creare的时间是否在30分钟内
            如果在就是正常支付

            如果不在就是超时支付

-----------------------
keyBy: oderId

session窗口: gap设置为 30分钟

如果窗口内有2个元素就一定是正常支付

如果是一个元素:
    有可能是create有可能是pay

    1. 如果是create, 把他存入大状态中

    2. 如果是pya, 判断状态中有没有create
        如果有create 就是超时支付

        如果没有create, 只有pay没有create
 */